BigQuery テーブル作成イベントで、description の日本語をカラム名にしたビューをバッチ処理で作成してみた。

from google.cloud import bigquery
import functions_framework
import os
import time

# INFORMATION_SCHEMAからカラムとdescriptionを取得するためのテンプレート
query_template = """
SELECT
  c.column_name,
  c.data_type,
  c.is_nullable,
  cf.description,
  c.ordinal_position,
  cf.field_path
FROM
  `{project_id}.{dataset_id}.INFORMATION_SCHEMA.COLUMNS` AS c
LEFT JOIN
  `{project_id}.{dataset_id}.INFORMATION_SCHEMA.COLUMN_FIELD_PATHS` AS cf
ON
  c.table_catalog = cf.table_catalog
  AND c.table_schema = cf.table_schema
  AND c.table_name = cf.table_name
  AND c.column_name = cf.column_name
WHERE
  c.table_name = '{table_id}'
ORDER BY
  c.ordinal_position
"""

# ビューのSQLを作成するためのテンプレート
view_sql_template = """
CREATE OR REPLACE VIEW `{project_id}.{dataset_id}.{view_id}` AS
SELECT
  {columns_sql}
FROM
  `{project_id}.{dataset_id}.{table_id}`
"""

def create_view(project_id, dataset_id, table_id):
    client = bigquery.Client()

    # カラム情報取得SQL作成
    query = query_template.format(project_id=project_id, dataset_id=dataset_id, table_id=table_id)
    # クエリを実行
    query_job = client.query(query)
    results = query_job.result()

    # カラム名とdescriptionを取得
    columns = []
    has_nested_structure = False
    missing_description = False
    for row in results:
        column_name = row['column_name']
        data_type = row['data_type']
        description = row['description']

        # ネスト構造を持つデータ型がある場合は終了
        if 'STRUCT' in data_type or 'RECORD' in data_type or 'ARRAY' in data_type or 'JSON' in data_type:
            has_nested_structure = True
            break

        # descriptionがないカラムがある場合は終了
        if not description:
            missing_description = True
            break

        columns.append(f"`{column_name}` AS `{description}`")

    # ネスト構造がある場合やdescriptionがないカラムがある場合は終了
    if has_nested_structure:
        print("[functions]The table contains nested structures. View creation aborted.")
        return
    elif missing_description:
        print("[functions]One or more columns are missing descriptions. View creation aborted.")
        return
    else:
        # CREATE VIEW SQL作成
        view_id = f"v_{table_id}"
        columns_sql = ",\n  ".join(columns)
        view_sql = view_sql_template.format(project_id=project_id, dataset_id=dataset_id, table_id=table_id, view_id=view_id, columns_sql=columns_sql)

        # ビューを作成
        client.query(view_sql).result()
        print(f"[functions]View `{view_id}` created successfully.")
        return

@functions_framework.cloud_event
def main(cloudevent):
    print(f"Event type: {cloudevent['type']}")
    if 'subject' in cloudevent:
        print(f"Subject: {cloudevent['subject']}")

    # 環境変数チェック
    PROJECT_ID = os.getenv('PROJECT_ID')
    SA_DATAFORM = os.getenv('SA_DATAFORM')
    if not PROJECT_ID or not SA_DATAFORM:
        raise Exception('Invalid env val.')

    # イベントデータチェック
    payload = cloudevent.data.get("protoPayload")
    if not payload:
        return
    principalEmail = payload.get('authenticationInfo', {}).get('principalEmail', None)
    if principalEmail != SA_DATAFORM:
        # Dataformからのクエリじゃない場合は処理せず終了
        return
    jobChange = payload.get('metadata', {}).get('jobChange', None)
    if not jobChange:
        return
    statementType = jobChange.get('job', {}).get('jobConfig', {}).get('queryConfig', {}).get('statementType', None)
    if statementType != 'CREATE_TABLE_AS_SELECT':
        return
    statementType = jobChange.get('job', {}).get('jobConfig', {}).get('queryConfig', {}).get('statementType', None)
    destinationTable = jobChange.get('job', {}).get('jobConfig', {}).get('queryConfig', {}).get('destinationTable', None)
    dst_dataset = destinationTable.split('/')[3]
    dst_table = destinationTable.split('/')[5]
    print(f"[functions] dst_dataset: {dst_dataset} / dst_table: {dst_table}")

    # TODO: すぐに実行するとdescription情報が取れないことがあったので、10秒待機
    time.sleep(10)

    create_view(PROJECT_ID, dst_dataset, dst_table)

functions-framework==3.*
google-cloud-bigquery>=3.25.0

gcloud functions deploy create_view \
    --gen2 \
    --region asia-northeast1 \
    --runtime python312 \
    --entry-point=main \
    --trigger-event-filters="type=google.cloud.audit.log.v1.written" \
    --trigger-event-filters="serviceName=bigquery.googleapis.com" \
    --trigger-event-filters="methodName=google.cloud.bigquery.v2.JobService.InsertJob" \
    --trigger-location=asia-northeast1 \
    --service-account=sa-functions@[PROJECT_ID].iam.gserviceaccount.com \
    --trigger-service-account=sa-functions@[PROJECT_ID].iam.gserviceaccount.com \
    --set-env-vars PROJECT_ID=[PROJECT_ID],SA_DATAFORM=service-[PROJECT_NO]@gcp-sa-dataform.iam.gserviceaccount.com